import com.zhenjun.domin.Item;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @author wangzj
 * @description SQL实现流式计算
 * @date 2020/7/14 0:50
 */
public class SqlStreamingDemo {
    public static void main(String[] args) throws Exception {
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance().useBlinkPlanner().inStreamingMode().build();
        StreamExecutionEnvironment strEvn = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment strTableEvn = StreamTableEnvironment.create(strEvn, settings);

        SingleOutputStreamOperator<Item> source = strEvn
                .addSource(new MyStreamingSource())
                .map(item -> item);

        DataStream<Item> evenSelect = source
                .split(item -> {
                    List<String> output = new ArrayList<>();
                    if (item.getId() % 2 == 0) {
                        output.add("even");
                    } else {
                        output.add("odd");
                    }
                    return output;
                }).select("even");

        DataStream<Item> oddSelect = source
                .split(item -> {
                    List<String> output = new ArrayList<>();
                    if (item.getId() % 2 == 0) {
                        output.add("even");
                    } else {
                        output.add("odd");
                    }
                    return output;
                }).select("odd");

        //注册成一张表，并指定表名和字段名
        strTableEvn.createTemporaryView("evenTable",evenSelect,"name,id");
        strTableEvn.createTemporaryView("oddTable",oddSelect,"name,id");

        Table queryTable = strTableEvn.sqlQuery("select a.id,a.name,b.id,b.name from evenTable as a join oddTable as b on a.name = b.name");
        queryTable.printSchema();

        strTableEvn.toRetractStream(queryTable, TypeInformation.of(new TypeHint<Tuple4<Integer,String,Integer,String>>(){})).print();

        strEvn.execute("Streaming SQL job");
    }
}
